package dws


import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import modes.AdvertisementInfo
import ods.SourceKafka
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ArrayBuffer

object AdvertisementStatisticsPer5Sec {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val eventLog: DataStream[String] = env.addSource(SourceKafka.getSource("stage9_event_log"))
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val record: DataStream[AdvertisementInfo] = eventLog.flatMap((x: String, collect: Collector[AdvertisementInfo]) => {
      val json: JSONObject = JSON.parseObject(x)
      val jsonArr: JSONArray = json.getJSONArray("lagou_event")
      val attr: JSONObject = json.getJSONObject("attr")
      val area: String = attr.getString("area")
      val userId: String = attr.getString("uid")
      jsonArr.forEach(x => {
        val jsonEvent: JSONObject = JSON.parseObject(x.toString)
        if (jsonEvent.getString("name").equals("ad")) {
          val jsonAd: JSONObject = jsonEvent.getJSONObject("json")
          val productId: String = jsonAd.getString("product_id")
          val timestamp: Long = jsonEvent.getLong("time")
          collect.collect(AdvertisementInfo(productId, timestamp, userId, area))
        }
      })
    })
    val keyed: KeyedStream[AdvertisementInfo, String] = record.filter(x => x.productId != null).assignAscendingTimestamps(x => x.timestamp).keyBy(x => x.productId)

    val windowed: WindowedStream[AdvertisementInfo, String, TimeWindow] = keyed.timeWindow(Time.hours(1), Time.seconds(5))
    val result: DataStream[(String, Int)] = windowed.aggregate(new AggregateFunction[AdvertisementInfo, Int, Int] {
      override def createAccumulator(): Int = 0

      override def add(value: AdvertisementInfo, accumulator: Int): Int = accumulator + 1

      override def getResult(accumulator: Int): Int = accumulator

      override def merge(a: Int, b: Int): Int = ???
    }, new WindowFunction[Int, (String, Int), String, TimeWindow] {
      override def apply(key: String, window: TimeWindow, input: Iterable[Int], out: Collector[(String, Int)]): Unit = {
        out.collect((s"$key:${window.getStart}-${window.getEnd}", input.iterator.next()))

      }
    })

    result.print()
    env.execute()
  }
}
